package com.dtp.core.thread;

import static com.dtp.common.constant.DynamicTpConst.TRACE_ID;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.MDC;

import com.dtp.common.entity.NotifyItem;
import com.dtp.core.notify.manager.NotifyHelper;
import com.dtp.core.reject.RejectHandlerGetter;
import com.dtp.core.spring.DtpLifecycleSupport;
import com.dtp.core.spring.SpringExecutor;
import com.dtp.core.support.runnable.DtpRunnable;
import com.dtp.core.support.runnable.NamedRunnable;
import com.dtp.core.support.wrapper.TaskWrapper;
import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;

/**
 * 动态线程池执行器
 *
 * @author yanhom
 * @since 1.0.0
 *
 * @remark muse
 **/
@Slf4j
public class DtpExecutor extends DtpLifecycleSupport implements SpringExecutor, ExecutorAdapter<ThreadPoolExecutor> {
    private String threadPoolAliasName;         // 线程池别名
    private String rejectHandlerName;           // 拒绝策略处理器名称
    private boolean notifyEnabled = true;       // 是否开启通知
    private List<NotifyItem> notifyItems;       // 通知项 {@link NotifyItemEnum}
    private List<String> platformIds;           // 通知平台id列表
    private List<TaskWrapper> taskWrappers = Lists.newArrayList();  // 任务包装器列表（do sth enhanced）
    private boolean preStartAllCoreThreads;                         // 是否提前开启所有核心线程，默认：false
    private long runTimeout;                                        // 任务执行超时时间（单位：ms）
    private long queueTimeout;                                      // 任务队列执行超时时间（单位：ms）
    private final LongAdder rejectCount = new LongAdder();          // 拒绝次数
    private final LongAdder runTimeoutCount = new LongAdder();      // 运行超时的任务数量
    private final LongAdder queueTimeoutCount = new LongAdder();    // 等待超时的队列数量

    /** -----------------------构造函数----------------------- */
    public DtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                       BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), new AbortPolicy());
    }
    public DtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                       BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new AbortPolicy());
    }
    public DtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                       BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
    }
    public DtpExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                       BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, RejectHandlerGetter.getProxy(handler));
        this.rejectHandlerName = handler.getClass().getSimpleName();
    }

    /** 执行任务 */
    @Override
    public void execute(Runnable task, long startTimeout) {execute(task);}
    @Override
    public void execute(Runnable command) {
        DtpRunnable dtpRunnable = (DtpRunnable) wrapTasks(command); // 执行任务包装器（如果存在任务包装器的话）
        dtpRunnable.startQueueTimeoutTask(this);                    // 开启超时任务队列
        super.execute(dtpRunnable);                                 // ThreadPoolExecutor.execute(...)
    }

    /** 执行前置任务 */
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);              // ThreadPoolExecutor.execute(...) doing nothing
        DtpRunnable runnable = (DtpRunnable) r;
        runnable.cancelQueueTimeoutTask();      // 取消超时任务队列
        runnable.startRunTimeoutTask(this, t);  // 开启运行超时任务
    }

    /** 执行后置任务 */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);                   // ThreadPoolExecutor.execute(...) doing nothing
        ((DtpRunnable) r).cancelRunTimeoutTask();   // 取消运行超时任务
        tryPrintError(r, t);                        // 输出错误信息（如果有的话）
        clearContext();                             // 清除上下文
    }

    /** 初始化方法 */
    @Override
    protected void initialize() {
        // 初始化消息通知
        NotifyHelper.initNotify(this);

        // 是否提前开启所有核心线程，默认：false，调用ThreadPoolExecutor.prestartAllCoreThreads()
        if (preStartAllCoreThreads) prestartAllCoreThreads();
    }

    /** 如果配置了任务包装器，则对command进行包装，最后创建DtpRunnable实例并进行返回 */
    protected Runnable wrapTasks(Runnable command) {
        if (CollectionUtils.isNotEmpty(taskWrappers))
            for (TaskWrapper t : taskWrappers) command = t.wrap(command);

        String taskName = (command instanceof NamedRunnable) ? ((NamedRunnable) command).getName() : null;
        command = new DtpRunnable(command, taskName);
        return command;
    }

    /** 清理MDC上下文 */
    private void clearContext() {MDC.remove(TRACE_ID);}

    /** 尝试打印错误信息 */
    private void tryPrintError(Runnable r, Throwable t) {
        if (Objects.nonNull(t)) {
            log.error("thread {} throw exception {}", Thread.currentThread(), t.getMessage(), t);
            return;
        }
        if (r instanceof FutureTask) {
            try {
                Future<?> future = (Future<?>) r;
                future.get();
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.error("thread {} throw exception {}", Thread.currentThread(), e.getMessage(), e);
            }
        }
    }

    /** 为了使Field可以通过反射来进行分配 */
    public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
        allowCoreThreadTimeOut(allowCoreThreadTimeOut); // 调用ThreadPoolExecutor#allowCoreThreadTimeOut
    }

    /** 获得队列容量 */
    public int getQueueCapacity() {
        int capacity = getQueue().size() + getQueue().remainingCapacity();
        return capacity < 0 ? Integer.MAX_VALUE : capacity;
    }

    /** ---------------------getter and setter--------------------- */
    @Override
    public ThreadPoolExecutor getOriginal() {return this;}
    public List<NotifyItem> getNotifyItems() {return notifyItems;}
    public void setNotifyItems(List<NotifyItem> notifyItems) {this.notifyItems = notifyItems;}
    public List<String> getPlatformIds() {return platformIds;}
    public void setPlatformIds(List<String> platformIds) {this.platformIds = platformIds;}
    public String getQueueName() {return getQueue().getClass().getSimpleName();}
    public String getRejectHandlerName() {return rejectHandlerName;}
    public void setRejectHandlerName(String rejectHandlerName) {this.rejectHandlerName = rejectHandlerName;}
    public void setTaskWrappers(List<TaskWrapper> taskWrappers) {this.taskWrappers = taskWrappers;}
    public void setPreStartAllCoreThreads(boolean preStartAllCoreThreads) {this.preStartAllCoreThreads = preStartAllCoreThreads;}
    public void incRejectCount(int count) {rejectCount.add(count);}
    public long getRejectCount() {return rejectCount.sum();}
    public void setRunTimeout(long runTimeout) {this.runTimeout = runTimeout;}
    public long getRunTimeout() {return runTimeout;}
    public LongAdder getRunTimeoutCount() {return runTimeoutCount;}
    public LongAdder getQueueTimeoutCount() {return queueTimeoutCount;}
    public void setQueueTimeout(long queueTimeout) {this.queueTimeout = queueTimeout;}
    public long getQueueTimeout() {return queueTimeout;}
    public String getThreadPoolAliasName() {return threadPoolAliasName;}
    public void setThreadPoolAliasName(String threadPoolAliasName) {this.threadPoolAliasName = threadPoolAliasName;}
    public boolean isNotifyEnabled() {return notifyEnabled;}
    public void setNotifyEnabled(boolean notifyEnabled) {this.notifyEnabled = notifyEnabled;}

}
